package udf.string;

import java.io.IOException;
import java.util.Iterator;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;


public class AddRank extends EvalFunc<DataBag> {

	private BagFactory bagFactory = BagFactory.getInstance();

	public DataBag exec(Tuple input) throws IOException {
		
		if (input.size() != 1 || input.get(0) == null) {
			return null;
		}

		else {
			int i = 1;
			DataBag bag = (DataBag) input.get(0);
			DataBag output = bagFactory.newDefaultBag();
			Iterator<Tuple> iterator = bag.iterator();

			while (iterator.hasNext()) {

				Tuple tup = iterator.next();

				Tuple temp = TupleFactory.getInstance().newTuple();
				for (int j = 0; j < tup.size(); j++) {

					temp.append(tup.get(j));
				}

				temp.append(i);
				i = i + 1;

				output.add(temp);

			}
			
			return output;

		}
	}

}
